package belf.migrate.engine.dao;

import belf.migrate.api.taskconf.SinkConf;
import belf.migrate.api.taskconf.SourceConf;
import belf.migrate.api.taskconf.TaskConf;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import lombok.extern.slf4j.Slf4j;

import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.util.HashMap;
import java.util.List;

/**
 * 从数据库中查询控制台配置的{@link belf.migrate.api.taskconf.TaskConf}信息
 */
@Slf4j
public class TaskConfDAO extends BaseDAO<TaskConf>{

    @Override
    public void save(TaskConf taskConf) {
        Connection connection = null;
        PreparedStatement ps = null;
        try {
            connection = getConnection();
            String sql = "INSERT INTO migration_task_conf" +
                    "(task_name,task_desc,task_type,source,sink,cdc,owner,create_time)" +
                    " values(?,?,?,?,?,?,?,?)";
            ps = connection.prepareStatement(sql);
            ps.setString(1, taskConf.getTaskName());
            ps.setString(2, taskConf.getTaskDesc());
            ps.setString(3, taskConf.getTaskType());
            ps.setString(4, taskConf.getSourceConf().toJson());
            ps.setString(5, taskConf.getSinkConf().toJson());
            ObjectMapper mapper = new ObjectMapper();
            ps.setString(6, mapper.writer().writeValueAsString(taskConf.getCdc()));
            ps.setString(7, taskConf.getOwner());
            ps.setDate(8, new java.sql.Date(taskConf.getCreateDate().getTime()));
            ps.executeUpdate();
            ps.close();
            log.info("Save TaskConf to db: {}", taskConf.getTaskName());
        } catch (SQLException e) {
            e.printStackTrace();
        } catch (JsonProcessingException e) {
            e.printStackTrace();
        } finally {
            try {
                if (null != ps) ps.close();
                if (null != connection) connection.close();
            } catch (SQLException e) {}
        }
    }

    @Override
    public TaskConf queryById(long id) {
        TaskConf taskConf = null;
        Connection connection = null;
        PreparedStatement ps = null;
        try {
            connection = getConnection();
            String sql = "SELECT * FROM migration_task_conf WHERE id=?";
            ps = connection.prepareStatement(sql);
            ps.setLong(1, id);
            ResultSet rs = ps.executeQuery();
            if (rs.next()) {
                ObjectMapper mapper = new ObjectMapper();
                taskConf = new TaskConf(rs.getInt("id"),
                        rs.getString("task_name"),
                        rs.getString("task_type"),
                        rs.getString("task_desc"),
                        mapper.readValue(rs.getString("source"), SourceConf.class),
                        mapper.readValue(rs.getString("sink"), SinkConf.class),
                        mapper.readValue(rs.getString("mapping"), HashMap.class),
                        mapper.readValue(rs.getString("cdc"), HashMap.class),
                        mapper.readValue(rs.getString("jdbc"), List.class),
                        rs.getString("owner"),
                        rs.getDate("create_time"));

            }
            rs.close();
        } catch (SQLException e) {
            e.printStackTrace();
        } catch (JsonProcessingException e) {
            e.printStackTrace();
        } finally {
            try {
                if (null != ps) ps.close();
                if (null != connection) connection.close();
            } catch (SQLException e) {}
        }
        return taskConf;
    }

    @Override
    public void deleteById(long id) {
        log.info("Delete TaskConf by id: {}", id);
        String sql = "DELETE FROM migration_task_conf where id=" + id;
        executeSql(sql);
    }
}
